package com.xrui.hbase.filter;

import com.google.common.base.Preconditions;
import com.xrui.hbase.DataRequest;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.hadoop.hbase.filter.Filter;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;

import java.io.IOException;

/**
 * A row filter that accepts or rejects rows at random.
 * <p>
 * <p> Wraps the HBase {@link org.apache.hadoop.hbase.filter.RandomRowFilter}. </p>
 * <p> A row is included if {@code random.nextFloat() < chance}. </p>
 */
public final class RandomRowFilter extends RowFilter {

    /**
     * The name of the chance node.
     */
    private static final String CHANCE_NODE = "chance";

    /**
     * Cutoff for random row selection: 0 means all excluded, 1 means all excluded.
     */
    private final float mChance;

    /**
     * Construct a filter that will return a row if {@code random.nextFloat() < chance}.
     *
     * @param chance Cutoff for random comparison selection.
     *               Chance must be between 0 an 1, inclusive: [0,1].
     *               0 means all excluded.
     *               1 means all included.
     */
    public RandomRowFilter(float chance) {
        Preconditions.checkArgument(chance >= 0 && chance <= 1, "chance must be between 0 and 1");
        this.mChance = chance;
    }

    /**
     * Get the row selection chance.
     *
     * @return the cutoff for random comparison selection.
     */
    public float getChance() {
        return this.mChance;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public DataRequest getDataRequest() {
        return DataRequest.builder().build();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter() throws IOException {
        return new org.apache.hadoop.hbase.filter.RandomRowFilter(this.mChance);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof RandomRowFilter)) {
            return false;
        } else {
            final RandomRowFilter otherRandomRowFilter = (RandomRowFilter) other;
            return new EqualsBuilder()
                .append(this.mChance, otherRandomRowFilter.mChance)
                .isEquals();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return new HashCodeBuilder()
            .append(this.mChance)
            .toHashCode();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected JsonNode toJsonNode() {
        ObjectNode rootNode = JsonNodeFactory.instance.objectNode();
        rootNode.put(CHANCE_NODE, this.mChance);
        return rootNode;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected Class<? extends RowFilterDeserializer> getDeserializerClass() {
        return RandomRowFilterDeserializer.class;
    }

    /**
     * A class to deserialize the RandomRowFilter from a JsonNode.
     */
    public static class RandomRowFilterDeserializer implements RowFilterDeserializer {
        /**
         * {@inheritDoc}
         */
        @Override
        public RowFilter createFromJson(JsonNode rootNode) {
            float chance = (float) rootNode.path(CHANCE_NODE).asDouble();
            return new RandomRowFilter(chance);
        }
    }
}
